package com.atguigu.Flink.state;

import com.atguigu.Flink.POJO.AppEvent;
import com.atguigu.Flink.POJO.Event;
import com.atguigu.Flink.POJO.ThirdPartyEvent;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;
//我们可以实现一个实时对账的需求，也就是app的支付操作和第三方的支付操作的一个双流Join。
//App的支付事件和第三方的支付事件将会互相等待5秒钟，如果等不来对应的支付事件，那么就输出报警信息。
public class Flink12_BillCheck {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<AppEvent> appDs = env.socketTextStream("hadoop102", 8888)
                .map(
                        line -> {
                            String[] split = line.split(",");
                            return new AppEvent(split[0], split[1], Long.valueOf(split[2]));
                        }
                ).filter(
                        event -> event.getEventType().equals("pay")
                ).assignTimestampsAndWatermarks(
                        WatermarkStrategy.<AppEvent>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner(
                                        new SerializableTimestampAssigner<AppEvent>() {
                                            @Override
                                            public long extractTimestamp(AppEvent appEvent, long l) {
                                                return appEvent.getTs();
                                            }
                                        }
                                )
                );

        SingleOutputStreamOperator<ThirdPartyEvent> thirdDs = env.socketTextStream("hadoop102", 9999)
                .map(
                        line -> {
                            String[] split = line.split(",");
                            return new ThirdPartyEvent(split[0], split[1],split[2], Long.valueOf(split[3]));
                        }
                ).filter(
                        event -> event.getEventType().equals("pay")
                ).assignTimestampsAndWatermarks(
                        WatermarkStrategy.<ThirdPartyEvent>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner(
                                        new SerializableTimestampAssigner<ThirdPartyEvent>() {
                                            @Override
                                            public long extractTimestamp(ThirdPartyEvent thirdDs, long l) {
                                                return thirdDs.getTs();
                                            }
                                        }
                                )
                );
        appDs.print();
        thirdDs.print();

        appDs.keyBy(
                k->k.getOrderId()
        ).connect(
                thirdDs.keyBy(
                        k->k.getOrderId()
                )
        ).process(
                new KeyedCoProcessFunction<String, AppEvent, ThirdPartyEvent, String>() {

                    private ValueState<AppEvent> appState;
                    private ValueState<ThirdPartyEvent> thirdState;
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //初始化状态
                        ValueStateDescriptor<AppEvent> app = new ValueStateDescriptor<>("app", AppEvent.class);
                        ValueStateDescriptor<ThirdPartyEvent> third = new ValueStateDescriptor<>("third", ThirdPartyEvent.class);
                        appState = getRuntimeContext().getState(app);
                        thirdState = getRuntimeContext().getState(third);

                    }

                    @Override
                    public void processElement1(AppEvent appEvent, KeyedCoProcessFunction<String, AppEvent, ThirdPartyEvent, String>.Context context, Collector<String> collector) throws Exception {
                        if(thirdState.value() != null){
                            collector.collect(appEvent.getOrderId() + "对账成功");
                            context.timerService().deleteEventTimeTimer(thirdState.value().getTs()+5000L);
                            thirdState.clear();
                        }else{
                            appState.update(appEvent);
                            context.timerService().registerEventTimeTimer(appEvent.getTs()+5000L);
                        }
                    }

                    @Override
                    public void processElement2(ThirdPartyEvent thirdPartyEvent, KeyedCoProcessFunction<String, AppEvent, ThirdPartyEvent, String>.Context context, Collector<String> collector) throws Exception {
                        if (appState.value() != null){
                            collector.collect(thirdPartyEvent.getOrderId()+"对账成功");
                            context.timerService().deleteEventTimeTimer(appState.value().getTs()+5000L);
                            appState.clear();
                        }else{
                            thirdState.update(thirdPartyEvent);
                            context.timerService().registerEventTimeTimer(thirdPartyEvent.getTs()+5000L);
                        }
                    }

                    @Override
                    public void onTimer(long timestamp, KeyedCoProcessFunction<String, AppEvent, ThirdPartyEvent, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
                        if (appState.value() != null){
                            out.collect("对账失败，第三方未到");
                        }
                        if (thirdState.value() != null){
                            out.collect("对账失败，app未到");
                        }
                        appState.clear();
                        thirdState.clear();
                    }
                }
        ).print();




        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
